use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{RwLock, broadcast};
use warp::ws::{Message, WebSocket};
use warp::Filter;
use futures::{SinkExt, StreamExt};
use serde_json;
use tracing::{info, error, debug, warn};
use uuid::Uuid;

use crate::api::models::{WebSocketMessage, DataPointModel, DeviceStatusModel};
use crate::storage::DataPoint;
use crate::error::{Result, ModbusCollectorError};

/// WebSocket连接管理器
#[derive(Debug)]
pub struct WebSocketManager {
    connections: Arc<RwLock<HashMap<String, WebSocketConnection>>>,
    broadcast_tx: broadcast::Sender<WebSocketMessage>,
}

/// WebSocket连接
#[derive(Debug)]
struct WebSocketConnection {
    id: String,
    client_info: ClientInfo,
}

/// 客户端信息
#[derive(Debug, Clone)]
struct ClientInfo {
    user_agent: Option<String>,
    remote_addr: Option<String>,
    connected_at: chrono::DateTime<chrono::Utc>,
}

impl WebSocketManager {
    pub fn new() -> Self {
        let (broadcast_tx, _) = broadcast::channel(1000);
        
        Self {
            connections: Arc::new(RwLock::new(HashMap::new())),
            broadcast_tx,
        }
    }
    
    /// 创建WebSocket过滤器
    pub fn websocket_filter(&self) -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
        let manager = self.clone();
        
        warp::path("ws")
            .and(warp::ws())
            .and(warp::header::optional::<String>("user-agent"))
            .and(warp::addr::remote())
            .map(move |ws: warp::ws::Ws, user_agent: Option<String>, remote_addr: Option<std::net::SocketAddr>| {
                let manager = manager.clone();
                ws.on_upgrade(move |socket| {
                    manager.handle_connection(socket, user_agent, remote_addr.map(|addr| addr.to_string()))
                })
            })
    }
    
    /// 处理WebSocket连接
    async fn handle_connection(
        &self,
        websocket: WebSocket,
        user_agent: Option<String>,
        remote_addr: Option<String>,
    ) {
        let connection_id = Uuid::new_v4().to_string();
        let client_info = ClientInfo {
            user_agent,
            remote_addr: remote_addr.clone(),
            connected_at: chrono::Utc::now(),
        };
        
        info!("New WebSocket connection: {} from {:?}", connection_id, remote_addr);
        
        // 注册连接
        {
            let mut connections = self.connections.write().await;
            connections.insert(connection_id.clone(), WebSocketConnection {
                id: connection_id.clone(),
                client_info: client_info.clone(),
            });
        }
        
        // 分割socket为发送和接收端
        let (mut ws_tx, mut ws_rx) = websocket.split();
        
        // 订阅广播频道
        let mut broadcast_rx = self.broadcast_tx.subscribe();
        let connections = self.connections.clone();
        let conn_id_for_cleanup = connection_id.clone();
        
        // 处理广播消息的任务
        let broadcast_task = tokio::spawn(async move {
            while let Ok(message) = broadcast_rx.recv().await {
                match serde_json::to_string(&message) {
                    Ok(json) => {
                        if let Err(e) = ws_tx.send(Message::text(json)).await {
                            debug!("Failed to send WebSocket message: {}", e);
                            break;
                        }
                    },
                    Err(e) => {
                        error!("Failed to serialize WebSocket message: {}", e);
                    }
                }
            }
        });
        
        // 处理客户端消息的任务
        let client_task = tokio::spawn(async move {
            while let Some(result) = ws_rx.next().await {
                match result {
                    Ok(message) => {
                        if message.is_text() {
                            if let Ok(text) = message.to_str() {
                                debug!("Received WebSocket message from {}: {}", connection_id, text);
                                // 这里可以处理客户端发送的消息
                                // 例如：订阅特定设备、认证等
                            }
                        } else if message.is_close() {
                            debug!("WebSocket connection {} closed by client", connection_id);
                            break;
                        }
                    },
                    Err(e) => {
                        warn!("WebSocket error on connection {}: {}", connection_id, e);
                        break;
                    }
                }
            }
        });
        
        // 等待任一任务完成
        tokio::select! {
            _ = broadcast_task => {
                debug!("Broadcast task ended for connection {}", conn_id_for_cleanup);
            },
            _ = client_task => {
                debug!("Client task ended for connection {}", conn_id_for_cleanup);
            },
        }
        
        // 清理连接
        {
            let mut connections = connections.write().await;
            connections.remove(&conn_id_for_cleanup);
        }
        
        info!("WebSocket connection {} disconnected", conn_id_for_cleanup);
    }
    
    /// 广播数据更新
    pub async fn broadcast_data_update(&self, data_point: &DataPoint) -> Result<()> {
        let message = WebSocketMessage::DataUpdate {
            data: DataPointModel::from(data_point.clone()),
        };
        
        if let Err(e) = self.broadcast_tx.send(message) {
            // 如果没有接收者，这是正常的
            debug!("No WebSocket subscribers for data update: {}", e);
        }
        
        Ok(())
    }
    
    /// 广播设备状态更新
    pub async fn broadcast_device_status(&self, status: &crate::storage::DeviceStatus) -> Result<()> {
        let message = WebSocketMessage::DeviceStatus {
            status: DeviceStatusModel::from(status.clone()),
        };
        
        if let Err(e) = self.broadcast_tx.send(message) {
            debug!("No WebSocket subscribers for device status: {}", e);
        }
        
        Ok(())
    }
    
    /// 广播系统状态
    pub async fn broadcast_system_status(&self, status: &crate::api::models::SystemStatus) -> Result<()> {
        let message = WebSocketMessage::SystemStatus {
            status: status.clone(),
        };
        
        if let Err(e) = self.broadcast_tx.send(message) {
            debug!("No WebSocket subscribers for system status: {}", e);
        }
        
        Ok(())
    }
    
    /// 广播错误消息
    pub async fn broadcast_error(&self, error_message: String) -> Result<()> {
        let message = WebSocketMessage::Error {
            message: error_message,
        };
        
        if let Err(e) = self.broadcast_tx.send(message) {
            debug!("No WebSocket subscribers for error: {}", e);
        }
        
        Ok(())
    }
    
    /// 获取连接统计信息
    pub async fn get_connection_stats(&self) -> WebSocketStats {
        let connections = self.connections.read().await;
        
        WebSocketStats {
            active_connections: connections.len(),
            total_connections_served: 0, // 可以添加计数器
        }
    }
    
    /// 获取所有连接信息
    pub async fn get_connections_info(&self) -> Vec<ConnectionInfo> {
        let connections = self.connections.read().await;
        
        connections.values()
            .map(|conn| ConnectionInfo {
                id: conn.id.clone(),
                user_agent: conn.client_info.user_agent.clone(),
                remote_addr: conn.client_info.remote_addr.clone(),
                connected_at: conn.client_info.connected_at,
                connected_duration_seconds: (chrono::Utc::now() - conn.client_info.connected_at).num_seconds() as u64,
            })
            .collect()
    }
}

impl Clone for WebSocketManager {
    fn clone(&self) -> Self {
        Self {
            connections: self.connections.clone(),
            broadcast_tx: self.broadcast_tx.clone(),
        }
    }
}

/// WebSocket统计信息
#[derive(Debug, serde::Serialize)]
pub struct WebSocketStats {
    pub active_connections: usize,
    pub total_connections_served: u64,
}

/// 连接信息
#[derive(Debug, serde::Serialize)]
pub struct ConnectionInfo {
    pub id: String,
    pub user_agent: Option<String>,
    pub remote_addr: Option<String>,
    pub connected_at: chrono::DateTime<chrono::Utc>,
    pub connected_duration_seconds: u64,
}